package cn.dmp.tags

import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.Row
import redis.clients.jedis.Jedis

object Tags4Business extends Tags{
    /**
      * 打标签的方法定义
      */
    override def makeTags(args: Any*): Map[String, Int] = {
        var map = Map[String, Int]()
        val row = args(0).asInstanceOf[Row]
        val jedis = args(1).asInstanceOf[Jedis]

        val lat = row.getAs[String]("lat")
        val longs = row.getAs[String]("long")

        if (StringUtils.isNotEmpty(lat) && StringUtils.isNotEmpty(longs)) {
            val lat2 = lat.toDouble
            val lon2 = longs.toDouble

            if(lat2 > 3 && lat2 < 54 && lon2 > 73 && lon2 < 136) {
                val geoHashCode = GeoHash.withCharacterPrecision(lat2, lon2, 8).toBase32
                val business = jedis.get(geoHashCode)
                if (StringUtils.isNotEmpty(business)) business.split(",").foreach(bs => map += "BS"+bs -> 1)
            }
        }
        map
    }
}
